package com.yb.kafka.consumer.listener;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @author yangbiao
 * @Description: 消费者---使用@KafkaListener注解,可以指定:主题,分区,消费组
 * @date 2018/11/6
 */
@Slf4j
@Service
@AllArgsConstructor
public class KafkaOneConsumer {
    private final StringRedisTemplate redisTemplate;


    /**
     * 这个简单点,只能获取到消息值
     *
     * @param message
     */
    @KafkaListener(topics = {"myMessage"}, groupId = "dd")
    public void receive(String message) {
        JSONObject jsonObject = JSONObject.parseObject(message);
        System.err.println("消费者One消费消息:" + jsonObject);
    }

    /**
     * ==========================需要开启手动ack也就是配置spring.kafka.listener.ack-mode=manual
     * 这个可以获取到除了消息的其他东西--->上面消费完了,下面就没得消费了,因为主题是一样的
     *==================================这个目前可以完成消费异常时消息不会丢失和不重复消费消息
     * @param record
     */
    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = {"myMessage"}, groupId = "bb")
    public void consumerMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        //默认设置可以手动提交offset
        boolean flag = true;
        try {
            if (Objects.nonNull(record)) {
                JSONObject jsonObject = JSONObject.parseObject(record.value());
                String id = jsonObject.getString("id");
                //判断消息是否被消费过
                if (!StringUtils.hasText(redisTemplate.opsForValue().get(id))) {
                    //注意业务事务回滚和消息之间是要分离的(消息异常后不提交offset,而业务这里要处理自己的回滚,包括redis的回滚)
                    try {
                        log.info("消费One的key:" + record.key());
                        log.info("消费One的value:" + record.value());
                        //消息消费过后,设置消息的id到redis,使用setIfAbsent,防止有多个相同的key存储
                        //但是使用set会造成重复很多key(通过redis处理消息重复消费的问题)
                        redisTemplate.opsForValue().setIfAbsent(id, id, 10, TimeUnit.DAYS);
                        //这个其实就是业务层面的异常,这里出现了异常,就不能让消息提交offset
                        //int i = 1 / 0;
                    } catch (Exception e) {
                        //删除设置的key,这样异常相当于回滚了
                        redisTemplate.delete(id);
                        //抛出异常,让消息无法提交offset
                        throw new IllegalArgumentException("");
                    }
                    //一定要控制好redis异常的问题,否则消息还是会丢失
                } else {
                    log.info(LocalDateTime.now() + ":已经消费了的消息id=" + id);
                }
            } else {
                log.info("获取到需要消费的消息为空");
            }
        } catch (Exception e) {
            //消费业务逻辑出现异常,不手动提交offset
            flag = false;
            log.info("consumer-one消费异常" + e.getMessage());
            //抛出异常,回滚事务(如果直接try掉,没有异常,事务不会回滚的)
            throw new IllegalArgumentException("我是consumer-one的异常");
        } finally {
            //处理offset的提交
            if (flag) {
                //手动提交offset(调用此方法意味着已经处理了分区中以前的所有消息)----->主动告知消息已经消费
                ack.acknowledge();
            }
        }
    }
}